package com.bilibili.juc.cf;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.RandomUtil;
import org.junit.Test;

import java.util.Date;
import java.util.Random;
import java.util.concurrent.*;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

/**
 * @auther zzyy
 * @create 2022-01-16 16:53
 */
public class CompletableFutureUseDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException {

        ExecutorService threadPool = Executors.newFixedThreadPool(3);

        try {
            CompletableFuture.supplyAsync(() -> {
                System.out.println(Thread.currentThread().getName() + "----come in");
                int result = ThreadLocalRandom.current().nextInt(10);
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("-----1秒钟后出结果：" + result);
                if (result > 2) {
                    int i = 10 / 0;
                }
                return result;
            }, threadPool).whenComplete((v, e) -> {
                if (e == null) {
                    System.out.println("-----计算完成，更新系统UpdateValue：" + v);
                }

            }).exceptionally(e -> {
                e.printStackTrace();
                System.out.println("异常情况：" + e.getCause() + "\t" + e.getMessage());
                return null;
            });

            System.out.println(Thread.currentThread().getName() + "线程先去忙其它任务");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            threadPool.shutdown();
        }


        //主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:暂停3秒钟线程
        //try { TimeUnit.SECONDS.sleep(3); } catch (InterruptedException e) { e.printStackTrace(); }

    }

    private static void future1() throws InterruptedException, ExecutionException {
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "----come in");
            int result = ThreadLocalRandom.current().nextInt(10);
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("-----1秒钟后出结果：" + result);
            return result;
        });

        System.out.println(Thread.currentThread().getName() + "线程先去忙其它任务");

        System.out.println(completableFuture.get());
    }

    @Test
    public void testCompletableFutureOfRunAsyncOrSupplyAsync() throws ExecutionException, InterruptedException {
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> System.out.println("hello!"));// 输出 "hello!"
        System.out.println("CompletableFuture.runAsync 没有返回值：" + future.get());
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "hello!");
        System.out.println("CompletableFuture.supplyAsync 有返回值：" + future2.get());
        assertEquals("hello!", future2.get());
    }

    @Test
    public void testCompletableFutureOfThenApply() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.completedFuture("hello!").thenApply(s -> s + "world!");
        System.out.println("future.completedFuture.thenApply=> " + future.get());
        assertEquals("hello!world!", future.get());
// 这次调用将被忽略。
        CompletableFuture<String> thenApplyResult = future.thenApply(s -> s + "nice!");
        System.out.println("thenApplyResult=> " + thenApplyResult.get());
        assertEquals("hello!world!", future.get());
        assertEquals("hello!world!nice!", thenApplyResult.get());
    }

    @Test
    public void testCompletableFutureOfThenApply2() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.completedFuture("hello!").thenApply(s -> s + "world!").thenApply(s -> s + "nice!");
        assertEquals("hello!world!nice!", future.get());
    }

    /*
    如果你不需要从回调函数中获取返回结果，可以使用 thenAccept() 或者 thenRun()。这两个方法的区别在于 thenRun() 不能访问异步计算的结果
     */
    @Test
    public void testCompletableFutureOfThenAcceptOrThenRun() throws ExecutionException, InterruptedException {
        CompletableFuture.completedFuture("hello!").thenApply(s -> s + "world!").thenApply(s -> s + "nice!").thenAccept(System.out::println);//hello!world!nice!

        CompletableFuture.completedFuture("hello!").thenApply(s -> s + "world!").thenApply(s -> s + "nice!").thenRun(() -> System.out.println("hello!"));//hello!
    }

    @Test
    public void testCompletableFutureOfExceptionally() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            if (true) {
                throw new RuntimeException("Computation error!");
            }
            return "hello!";
        }).exceptionally(ex -> {
            System.out.println(ex.toString());// CompletionException
            return "world!";
        });
        System.out.println("result==>> " + future.get());
        assertEquals("world!", future.get());
    }

    @Test
    public void testCompletableFutureOfExceptionally2() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            if (true) {
                throw new RuntimeException("Computation error!");
            }
            return "hello!";
        }).exceptionally(ex -> {
            System.out.println(ex.toString());// CompletionException
            throw new RuntimeException(ex);
        });
        System.out.println("result==>> " + future.get());
        assertEquals("world!", future.get());
    }

    @Test
    public void testCompletableFutureOfCompleteExceptionally() throws ExecutionException, InterruptedException {
        CompletableFuture<String> completableFuture = new CompletableFuture<>();
        completableFuture.completeExceptionally(new RuntimeException("Calculation failed!"));
        // completableFuture.complete("kkk");
        System.out.println(completableFuture.get());
        ; // ExecutionException
    }

    /*
    组合 CompletableFuture

你可以使用 thenCompose() 按顺序链接两个 CompletableFuture 对象，实现异步的任务链。
它的作用是将前一个任务的返回结果作为下一个任务的输入参数，从而形成一个依赖关系
     */
    @Test
    public void testCompletableFutureOfThenCompose() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "hello!").thenCompose(s -> CompletableFuture.supplyAsync(() -> s + "world!"));
        assertEquals("hello!world!", future.get());
        System.out.println(future.get());
    }

    /*
    先执行A 将A执行的结果再传递给B执行
     */
    @Test
    public void testCompletableFutureOfThenCompose2() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t -----A come in " + DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + "\t -----A over    " + DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));
            return "hello!";
        }).thenCompose(s -> CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t -----B come in " + DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + "\t -----B over    " + DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));
            return s + "world!";
        }));
        assertEquals("hello!world!", future.get());
        System.out.println(future.get());
 /*
 ForkJoinPool.commonPool-worker-1	 -----A come in 2024-11-19 13:44:33
ForkJoinPool.commonPool-worker-1	 -----A over    2024-11-19 13:44:35
ForkJoinPool.commonPool-worker-1	 -----B come in 2024-11-19 13:44:35
ForkJoinPool.commonPool-worker-1	 -----B over    2024-11-19 13:44:36
hello!world!
  */
    }

    /*
    A,B一起执行当它们都执行完后再执行BiFunction函数，也是将它们的执行的结果再进行处理
     */
    @Test
    public void testCompletableFutureOfThenCombine() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t -----A come in " + DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + "\t -----A over    " + DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));
            return "hello!";
        }).thenCombine(CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName() + "\t -----B come in " + DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + "\t -----B over    " + DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));
            return "world!";
        }), (x, y) -> x + y);
        assertEquals("hello!world!", future.get());
        System.out.println(future.get());
/*
ForkJoinPool.commonPool-worker-1	 -----A come in 2024-11-19 13:49:36
ForkJoinPool.commonPool-worker-2	 -----B come in 2024-11-19 13:49:36
ForkJoinPool.commonPool-worker-2	 -----B over    2024-11-19 13:49:37
ForkJoinPool.commonPool-worker-1	 -----A over    2024-11-19 13:49:38
hello!world!
 */
    }
/*
那 thenCompose() 和 thenCombine() 有什么区别呢？

    thenCompose() 可以链接两个 CompletableFuture 对象，并将前一个任务的返回结果作为下一个任务的参数，它们之间存在着先后顺序。
    thenCombine() 会在两个任务都执行完成后，把两个任务的结果合并。两个任务是并行执行的，它们之间并没有先后依赖顺序。
————————————————

                            版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。

原文链接：https://blog.csdn.net/Pre_W/article/details/143750030
 */


    /*
    例如，如果我们想要实现 task1 和 task2 中的任意一个任务执行完后就执行 task3 的话，可以使用 acceptEither()
     */
    @Test
    public void testCompletableFutureOfAcceptEitherAsync() throws ExecutionException, InterruptedException {
        CompletableFuture<String> task = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务1开始执行，当前时间：" + System.currentTimeMillis());
            try {
                long sleepTime = RandomUtil.randomInt(1000, 6000);
                System.out.println("任务1 sleepTime=> " + sleepTime);
                Thread.sleep(sleepTime);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("任务1执行完毕，当前时间：" + System.currentTimeMillis());
            return "task1";
        });

        CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> {
            System.out.println("任务2开始执行，当前时间：" + System.currentTimeMillis());
            try {
                long sleepTime = RandomUtil.randomInt(1000, 5000);
                System.out.println("任务2 sleepTime=> " + sleepTime);
                Thread.sleep(sleepTime);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("任务2执行完毕，当前时间：" + System.currentTimeMillis());
            return "task2";
        });

        CompletableFuture<Void> eitherAsync = task.acceptEitherAsync(task2, (res) -> {
            System.out.println("任务3开始执行，当前时间：" + System.currentTimeMillis());
            System.out.println("上一个任务的结果为：" + res);
        });
        System.out.println("任务执行完了" + eitherAsync.get());

// 增加一些延迟时间，确保异步任务有足够的时间完成
//        try {
//            Thread.sleep(2000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
    }

    /*
并行运行多个 CompletableFuture

你可以通过 CompletableFuture 的 allOf()这个静态方法来并行运行多个 CompletableFuture

实际项目中，我们经常需要并行运行多个互不相关的任务，这些任务之间没有依赖关系，可以互相独立地运行

比说我们要读取处理 6 个文件，这 6 个任务都是没有执行顺序依赖的任务，但是我们需要返回给用户的时候将这几个文件的处理的结果进行统计整理。像这种情况我们就可以使用并行运行多个 CompletableFuture 来处理
————————————————

                            版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。

原文链接：https://blog.csdn.net/Pre_W/article/details/143750030
 */
    @Test
    public void testCompletableFutureOfAllOf() throws ExecutionException, InterruptedException {
        Random rand = new Random();
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println(Thread.currentThread().getName() + "\t -----A come in " + DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));
                Thread.sleep(1000 + rand.nextInt(1000));
                System.out.println(Thread.currentThread().getName() + "\t -----A over " + DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                System.out.println("future1 done...");
            }
            return "abc";
        });
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println(Thread.currentThread().getName() + "\t -----B come in " + DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));
                Thread.sleep(1000 + rand.nextInt(1000));
                System.out.println(Thread.currentThread().getName() + "\t -----B over " + DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                System.out.println("future2 done...");
            }
            return "efg";
        });
        CompletableFuture<Void> completableFuture = CompletableFuture.allOf(future1, future2);
        completableFuture.thenRun(() -> {
            System.out.println("all futures done..." + DateUtil.format(new Date(), DatePattern.NORM_DATETIME_PATTERN));
        });
        completableFuture.join();
        assertTrue(completableFuture.isDone());
        System.out.println("future1==>> " + future1.get());
        System.out.println("future2==>> " + future2.get());

    }

    /*
    CompletableFuture 使用建议
使用自定义线程池

我们上面的代码示例中，为了方便，都没有选择自定义线程池。实际项目中，这是不可取的

CompletableFuture 默认使用全局共享的 ForkJoinPool.commonPool() 作为执行器，所有未指定执行器的异步任务都会使用该线程池。这意味着应用程序、多个库或框架（如 Spring、第三方库）若都依赖 CompletableFuture，默认情况下它们都会共享同一个线程池

虽然 ForkJoinPool 效率很高，但当同时提交大量任务时，可能会导致资源竞争和线程饥饿，进而影响系统性能

为避免这些问题，建议为 CompletableFuture 提供自定义线程池，带来以下优势：

    隔离性：为不同任务分配独立的线程池，避免全局线程池资源争夺。
    资源控制：根据任务特性调整线程池大小和队列类型，优化性能表现。
    异常处理：通过自定义 ThreadFactory 更好地处理线程中的异常情况。
————————————————

                            版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。

原文链接：https://blog.csdn.net/Pre_W/article/details/143750030
尽量避免使用 get()

CompletableFuture的get()方法是阻塞的，尽量避免使用。如果必须要使用的话，需要添加超时时间，否则可能会导致主线程一直等待，无法执行其他任务
     */
    @Test
    public void testCompletableFutureOfAvoidUsingGetMethod() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(10_000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Hello, world!";
        });

        // 获取异步任务的返回值，设置超时时间为 5 秒
        try {
            String result = future.get(5, TimeUnit.SECONDS);
            System.out.println(result);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            // 处理异常
            e.printStackTrace();
        }

    }

    /*
    方法：

    runAsync 无返回值
    supplyAsync 有返回值
     */
    @Test
    public void testCompletableFutureOfRunAsyncAndSupplyAsync() throws ExecutionException, InterruptedException {
        ExecutorService threadPool = Executors.newFixedThreadPool(3);
//无返回值
        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            try {
                TimeUnit.MICROSECONDS.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, threadPool);//不指定线程池就会使用默认的线程池
        System.out.println("runAsync 无返回值=> " + future.get());//null
//有返回值
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread().getName());
            try {
                TimeUnit.MICROSECONDS.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "supplyAsyncReturnValue999";
        }, threadPool);
        System.out.println("supplyAsync 有返回值=> " + future1.get());
        threadPool.shutdown();
    }

    @Test
    public void testCompletableFutureOfWhenCompleteAndExceptionally() throws ExecutionException, InterruptedException {
        ExecutorService threadPool = Executors.newFixedThreadPool(3);
        CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(() -> {
            int nextInt = new Random().nextInt(10);
                        int a =  10 /0;
            return nextInt;
        }, threadPool).whenComplete((v, e) -> {//获得上一步执行完返回的结果 v；可能出现的异常 e
            if (e == null) {
                System.out.println("成功获得结果：" + v);
            }
        }).exceptionally(e -> {//发生异常后自动调用
            e.printStackTrace();
            System.out.println("发生异常：" + e.getMessage());
            return null;
        });

        if (completableFuture.isDone()){
            System.out.println("结果: " + completableFuture.get());
        }

        threadPool.shutdown();
        //todo 主线程执行任务 （注意：主线程结束后默认线程池会自动关闭，推荐使用自定义线程池）
    }

    /*
    handle：和thenApply类似，但发生异常时后续操作可以正常执行
     */
    @Test
    public void testCompletableFutureOfWhenCompleteAndExceptionally2() throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> handleFuture = CompletableFuture.supplyAsync(() -> {
            System.out.println(1);
            return 1;
        }).handle((v, e) -> {//第一步发生异常停止运行，但后面可以正常运行，直至最后把异常打印出来
            int i = 10 / 0;
            System.out.println(3);
            return v + 2;
        }).handle((v, e) -> {//这里正常输出
            System.out.println(6);
            return v + 3;
        }).whenComplete((v, e) -> {
            if (e == null) {
                System.out.println("handle：" + v);
            }
        }).exceptionally(e -> {
            e.printStackTrace();
            return null;
        });
        System.out.println("result==>> "+handleFuture.get());
/*
1
6
java.util.concurrent.CompletionException: java.lang.NullPointerException: Cannot invoke "java.lang.Integer.intValue()" because "v" is null
	at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:315)
	at java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:320)
	at java.base/java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:936)
	at java.base/java.util.concurrent.CompletableFuture.uniHandleStage(CompletableFuture.java:950)
	at java.base/java.util.concurrent.CompletableFuture.handle(CompletableFuture.java:2340)
	at com.bilibili.juc.cf.CompletableFutureUseDemo.testCompletableFutureOfWhenCompleteAndExceptionally2(CompletableFutureUseDemo.java:461)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:568)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:59)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:56)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
	at org.junit.runners.BlockJUnit4ClassRunner$1.evaluate(BlockJUnit4ClassRunner.java:100)
	at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:366)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:103)
	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:63)
	at org.junit.runners.ParentRunner$4.run(ParentRunner.java:331)
	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:79)
	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:329)
	at org.junit.runners.ParentRunner.access$100(ParentRunner.java:66)
	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:293)
	at org.junit.runners.ParentRunner$3.evaluate(ParentRunner.java:306)
	at org.junit.runners.ParentRunner.run(ParentRunner.java:413)
	at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
	at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:69)
	at com.intellij.rt.junit.IdeaTestRunner$Repeater$1.execute(IdeaTestRunner.java:38)
	at com.intellij.rt.execution.junit.TestsRepeater.repeat(TestsRepeater.java:11)
	at com.intellij.rt.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:35)
	at com.intellij.rt.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:235)
	at com.intellij.rt.junit.JUnitStarter.main(JUnitStarter.java:54)
Caused by: java.lang.NullPointerException: Cannot invoke "java.lang.Integer.intValue()" because "v" is null
	at com.bilibili.juc.cf.CompletableFutureUseDemo.lambda$testCompletableFutureOfWhenCompleteAndExceptionally2$42(CompletableFutureUseDemo.java:463)
	at java.base/java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:934)
	... 30 more
result==>> null
 */
    }


}
